package com.mqstudy.activemq.consumer;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.mqstudy.common.model.Message;
import com.mqstudy.activemq.config.ActiveMQConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.annotation.JmsListener;
import org.springframework.stereotype.Component;

import java.util.concurrent.atomic.AtomicLong;

/**
 * ActiveMQ消费者
 */
@Slf4j
@Component
public class ActiveMQConsumer {
    
    @Autowired
    private ObjectMapper objectMapper;
    
    private final AtomicLong messageCount = new AtomicLong(0);
    
    /**
     * 监听测试队列消息
     */
    @JmsListener(destination = ActiveMQConfig.TEST_QUEUE, containerFactory = "queueListenerFactory")
    public void handleQueueMessage(String messageBody) {
        try {
            log.info("收到队列消息: {}", messageBody);
            
            Message message = objectMapper.readValue(messageBody, Message.class);
            processMessage(message);
            
            long count = messageCount.incrementAndGet();
            log.info("队列消息处理完成，总处理数量: {}", count);
            
        } catch (Exception e) {
            log.error("处理队列消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 监听测试主题消息
     */
    @JmsListener(destination = ActiveMQConfig.TEST_TOPIC, containerFactory = "topicListenerFactory")
    public void handleTopicMessage(String messageBody) {
        try {
            log.info("收到主题消息: {}", messageBody);
            
            Message message = objectMapper.readValue(messageBody, Message.class);
            processMessage(message);
            
            long count = messageCount.incrementAndGet();
            log.info("主题消息处理完成，总处理数量: {}", count);
            
        } catch (Exception e) {
            log.error("处理主题消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 监听订单队列消息
     */
    @JmsListener(destination = ActiveMQConfig.ORDER_QUEUE, containerFactory = "queueListenerFactory")
    public void handleOrderMessage(String messageBody) {
        try {
            log.info("收到订单消息: {}", messageBody);
            
            Message message = objectMapper.readValue(messageBody, Message.class);
            processOrderMessage(message);
            
            log.info("订单消息处理完成");
            
        } catch (Exception e) {
            log.error("处理订单消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 监听用户主题消息
     */
    @JmsListener(destination = ActiveMQConfig.USER_TOPIC, containerFactory = "topicListenerFactory")
    public void handleUserMessage(String messageBody) {
        try {
            log.info("收到用户消息: {}", messageBody);
            
            Message message = objectMapper.readValue(messageBody, Message.class);
            processUserMessage(message);
            
            log.info("用户消息处理完成");
            
        } catch (Exception e) {
            log.error("处理用户消息失败: {}", e.getMessage(), e);
        }
    }
    
    /**
     * 处理通用消息
     */
    private void processMessage(Message message) {
        log.info("开始处理消息: {}", message.getMessageId());
        
        // 模拟业务处理时间
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 处理订单消息
     */
    private void processOrderMessage(Message message) {
        log.info("开始处理订单消息: {}", message.getMessageId());
        
        // 模拟订单处理逻辑
        try {
            Thread.sleep(150);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("订单消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 处理用户消息
     */
    private void processUserMessage(Message message) {
        log.info("开始处理用户消息: {}", message.getMessageId());
        
        // 模拟用户处理逻辑
        try {
            Thread.sleep(80);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        
        log.info("用户消息处理完成: {}", message.getMessageId());
    }
    
    /**
     * 获取消息处理统计
     */
    public long getMessageCount() {
        return messageCount.get();
    }
} 